package com.kai.mqtt.protocol;

import com.kai.mqtt.constant.AttributeKeyConstant;
import com.kai.mqtt.message.service.ISubscribeMessageService;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author weijk
 * @since 2024/6/3
 */
@Slf4j
@Component
public class Unsubscribe extends AbstractProtocol {

    @Resource
    private ISubscribeMessageService subscribeMessageService;

    @Override
    public void process(Channel channel, MqttMessage message) {
        MqttUnsubscribeMessage msg = (MqttUnsubscribeMessage) message;
        List<String> topicFilters = msg.payload().topics();
        String clientId = channel.attr(AttributeKeyConstant.CLIENT_ID).get();
        topicFilters.forEach(topicFilter -> {
            subscribeMessageService.remove(topicFilter, clientId);
            log.info("receive - unsubscribe, clientId: {}, topicFilter: {}", clientId, topicFilter);
        });
        MqttUnsubAckMessage unsubAckMessage = (MqttUnsubAckMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                MqttMessageIdVariableHeader.from(msg.variableHeader().messageId()), null);
        log.debug("send - unsubAck, clientId: {}, messageId: {}", clientId, msg.variableHeader().messageId());
        channel.writeAndFlush(unsubAckMessage);
    }
}
